package com.goldman.message.common.event;


import com.alibaba.fastjson.JSON;
import com.goldman.message.common.event.annotation.InputEvent;
import com.goldman.message.common.event.config.InputEventProp;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.tuple.Pair;
import org.aspectj.lang.JoinPoint;
import org.aspectj.lang.ProceedingJoinPoint;
import org.aspectj.lang.annotation.AfterThrowing;
import org.aspectj.lang.annotation.Around;
import org.aspectj.lang.annotation.Aspect;
import org.aspectj.lang.annotation.Pointcut;
import org.redisson.api.RLock;
import org.redisson.api.RMapCache;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
import org.springframework.stereotype.Component;

import java.util.concurrent.TimeUnit;

@Slf4j
@Aspect
@Component
@ConditionalOnClass(InputEvent.class)
public class InputEventAspect {

    @Autowired
    private RedissonClient redisson;
    @Autowired
    private InputEventProp inputEventProp;

    private static final String MESSAGE_REDIS_KEY = "MSG_ID";

    /**
     * 定义切点
     */
    @Pointcut("@within(com.goldman.message.common.event.annotation.InputEvent)")
    public void inputEventMethod() {
    }

    /**
     * 消息重复校验环绕通知
     *
     * @param joinPoint
     * @throws Throwable
     */
    @Around("inputEventMethod()")
    public Object procceed(ProceedingJoinPoint joinPoint) throws Throwable {
        Object[] objs = joinPoint.getArgs();
        if (null == objs || objs.length == 0) {
            log.warn("请求的入参对象为空,直接丢弃消息");
            return MQError.MESSAGE_EMPTY.getMsg();
        }
        Object obj = objs[0];
        Pair<MessageModel, String> pair = parseJson(obj);
        log.debug("mq请求报文:[{}]", pair.getRight());
        if (StringUtils.isBlank(pair.getLeft().getMsgId())) {
            log.warn("msgId为空,直接丢弃消息,请求报文为:[{}]", pair.getRight());
            return MQError.MESSAGE_ID_EMPTY.getMsg();
        }
        // 在redis中设置redisKey,防止msgId重复提交
        boolean flag = checkRedisMsgId(pair.getLeft(), pair.getRight());
        if (!flag) {
            return MQError.DUPLICATED_REQUEST.getMsg();
        }
        return joinPoint.proceed();
    }

    private Pair<MessageModel, String> parseJson(Object obj) {
        String objJsonStr = JSON.toJSONString(obj);
        MessageModel messageModel = JSON.parseObject(objJsonStr, MessageModel.class);
        return Pair.of(messageModel, objJsonStr);
    }


    @AfterThrowing("inputEventMethod()")
    public void AfterThrowing(JoinPoint joinPoint) {
        Object[] objs = joinPoint.getArgs();
        Pair<MessageModel, String> paris = parseJson(objs[0]);
        delRedisMsgId(paris.getLeft(), paris.getRight());
    }

    private String getMessageRedisKey(String messageId, String type) {
        return String.format("%s:%s:%s", MESSAGE_REDIS_KEY, messageId, type);
    }

    /**
     * @MethodName: checkRedisMsgId
     * @Description: 在redis中设置redisKey, 防止msgId重复提交
     * @Param: [json]
     * @Return: boolean
     * @Author: wpp
     * @Date: 2020/1/1 15:26
     **/
    private boolean checkRedisMsgId(MessageModel messageModel, String jsonStr) {
        String lockKey = getMessageRedisKey(messageModel.getMsgId(), messageModel.getType());
        RLock disLock = redisson.getLock(lockKey);
        boolean isLock;
        try {
            // 尝试获取分布式锁
            isLock = disLock.tryLock(inputEventProp.getLockWaitTime(), inputEventProp.getLockExpireTime(), TimeUnit.SECONDS);
            if (isLock) {
                // 校验通过,获取到锁
                return putRedisKey(jsonStr, messageModel.getMsgId(), messageModel.getType());
            } else {
                log.warn("msgId:[{}]重复交易,直接丢弃消息,请求报文为:[{}]", messageModel.getMsgId(), jsonStr);
                return false;
            }
        } catch (Exception e) {
            log.error("获取锁失败,请求报文:[{}]", jsonStr);
            throw new RuntimeException("message process, get redis lock error");
        } finally {
            if (disLock.isLocked()) {
                if (disLock.isLocked() && disLock.isHeldByCurrentThread()) {
                    disLock.unlock();
                }

            }

        }
    }

    /**
     * @MethodName: delRedisMsgId
     * @Description: 如果消息消费异常, 删除msgId在redis的记录, 以便于再一次消费
     * @Param: [json]
     * @Return: boolean
     * @Author: wpp
     * @Date: 2020/1/1 15:26
     **/
    private boolean delRedisMsgId(MessageModel messageModel, String jsonStr) {
        String msgId = messageModel.getMsgId();
        String type = messageModel.getType();
        RLock disLock = redisson.getLock(msgId);
        boolean isLock;
        try {
            // 尝试获取分布式锁
            isLock = disLock.tryLock(inputEventProp.getLockWaitTime(), inputEventProp.getExpireTime(), TimeUnit.SECONDS);
            if (isLock) {
                // 校验通过,获取到锁
                return delRedisKey(jsonStr, msgId, type);
            } else {
                log.warn("msgId:[{}]重复交易,直接丢弃消息,请求报文为:[{}]", msgId, jsonStr);
                return false;
            }
        } catch (Exception e) {
            log.error("获取锁失败,请求报文:[{}]", jsonStr);
            throw new RuntimeException("message process, get redis lock error", e);
        } finally {
            // 无论如何, 最后都要解锁
            if (disLock.isLocked() && disLock.isHeldByCurrentThread()) {
                disLock.unlock();
            }
        }
    }

    /**
     * @MethodName: putRedisKey
     * @Description: 往redis设置数据
     * @Param: [json, msgId]
     * @Return: boolean
     * @Author: wpp
     * @Date: 2020/1/2 19:51
     **/
    private boolean putRedisKey(String jsonStr, String msgId, String type) {
        String msgIdKey = getMessageRedisKey(msgId, type);
        RMapCache<String, String> rMap = redisson.getMapCache(msgIdKey);
        String value = rMap.get(msgIdKey);
        // 如果msgId之前使用过,视为重复交易
        if (StringUtils.isNotBlank(value)) {
            log.warn("msgId:[{}], type:[{}]重复交易,直接丢弃消息,请求报文为:[{}]",
                    msgId, type, jsonStr);
            return false;
        }
        // 这里要先put 再expire，否则expire不生效
        rMap.put(msgIdKey, jsonStr);
        rMap.expire(inputEventProp.getExpireTime(), TimeUnit.SECONDS);
        return true;
    }

    /**
     * @MethodName: delRedisKey
     * @Description: 删除redis数据
     * @Param: [json, msgId]
     * @Return: boolean
     * @Author: wpp
     * @Date: 2020/1/2 19:52
     **/
    private boolean delRedisKey(String jsonStr, String msgId, String type) {
        String msgIdKey = getMessageRedisKey(msgId, type);
        RMapCache<String, String> rMap = redisson.getMapCache(msgIdKey);
        String value = rMap.get(msgIdKey);
        // 如果redis不存在,直接返回
        if (StringUtils.isBlank(value)) {
            log.warn("value不存在:[{}]", jsonStr);
            return false;
        }
        return rMap.delete();
    }
}
